package ldh.im.rpc.connection.handle;

import ldh.im.rpc.config.RpcConfig;
import ldh.im.rpc.config.RpcMapperInfo;
import ldh.im.rpc.connection.RpcDataChannel;
import ldh.im.rpc.connection.executor.RpcRouterExecutor;
import ldh.im.rpc.dto.RpcData;
import ldh.im.rpc.dto.RpcExtData;
import ldh.im.rpc.exception.ExceptionResp;
import ldh.im.rpc.exception.RpcErrorCode;
import ldh.im.rpc.exception.RpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class RouterHandlerService implements HandlerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(RouterHandlerService.class);

    private RpcRouterExecutor rpcRouterExecutor = null;
    private RpcConfig rpcConfig;

    public RouterHandlerService(RpcConfig rpcConfig, RpcRouterExecutor rpcRouterExecutor) {
        this.rpcConfig = rpcConfig;
        this.rpcRouterExecutor = rpcRouterExecutor;
    }

    @Override
    public void handle(RpcDataChannel channel, RpcData rpcData) throws RpcException {
        long time = System.currentTimeMillis();
        String path = rpcData.getPath();
        LOGGER.info("handle start {}", path);
        try {
            RpcSession rpcSession = rpcConfig.getRpcCacheService().getSession(channel);
            Object dto = rpcData.getData();
            List<RpcFilter> imFilters = rpcConfig.getAllFilters();

            RpcContext rpcContext = new DefaultRpcContext(rpcConfig, rpcSession, rpcData);
            for(RpcFilter imFilter : imFilters) {
                if (!imFilter.before(rpcContext, rpcData)) {
                    LOGGER.error("被处理器拦截");
                    ExceptionResp exceptionResp = new ExceptionResp(RpcErrorCode.Filter_failure.getErrorCode(), imFilter.getClass().getName());
                    writeErrorResult(channel, rpcData, exceptionResp);
                    return;
                }
            }
            RpcMapperInfo rpcMapperInfo = rpcConfig.getRouter(rpcData.getPath());
            Object resultObj = rpcRouterExecutor.executor(rpcData.getPath(), rpcContext, dto);
            LOGGER.info("handle result {}", path);
            if (rpcContext.get(RpcContext.ASYN_HANDLE) != null) {  // 内部异步处理
                if (resultObj != null) {
                    writeResult(channel, rpcData, resultObj);
                }
            } else if (rpcMapperInfo.getMapper().reply()) {
                writeResult(channel, rpcData, resultObj);
            }
        } catch (RpcException e) {
            LOGGER.error("业务处理失败1, error {}", e);
            String msg = e.getErrorCode().getErrorMessage();
            Throwable throwable = e;
            while(throwable.getCause() != null) {
                throwable = throwable.getCause();
            }
            if (e.getErrorCode() == RpcErrorCode.Business_error) {
                msg = "服务端错误";
            } else if (e.getErrorCode().getErrorMessage().contains("%")) {
                msg = throwable.getMessage();
            } else {
                LOGGER.error("业务处理失败1", e);
                if (e.getErrorCode().getErrorMessage().contains("%")) {
                    msg = e.getMessage();
                }
            }

            ExceptionResp exceptionResp = new ExceptionResp(e.getErrorCode().getErrorCode(), msg);
            writeErrorResult(channel, rpcData, exceptionResp);
        } catch (Exception e) {
            LOGGER.error("业务处理失败2, error {}", e);
            ExceptionResp exceptionResp = new ExceptionResp(RpcErrorCode.Business_error.getErrorCode(), e.getMessage());
            writeErrorResult(channel, rpcData, exceptionResp);
        } finally {
            long consumeTime = (System.currentTimeMillis()-time);
            if (consumeTime > 500) { // 大约0.5s，大约warn 日志
                LOGGER.warn("===============handle {} warn too time {}", path, consumeTime);
            }
            LOGGER.info("handle {} message time {}", path, consumeTime);
        }
    }

    @Override
    public void asyncHandle(RpcDataChannel channel, RpcData rpcData) throws RpcException {
        rpcConfig.getBusinessExecutorService().submit(()->{
            handle(channel, rpcData);
        });
    }

    private void writeResult(RpcDataChannel channel, RpcData rpcData, Object resultObj) {
        RpcData resultRpcData = buildRpcData(rpcData, 100, 1, 0, resultObj);
        channel.write(resultRpcData);
    }

    private void writeErrorResult(RpcDataChannel channel, RpcData rpcData, ExceptionResp resultObj) {
        RpcData resultRpcData = buildRpcData(rpcData, -1, 10, 0, resultObj);
        channel.write(resultRpcData);
    }

    private RpcData buildRpcData(RpcData rpcData, int status, int version, int authCode, Object resultObj) {
        RpcData resultRpcData = new RpcData();
        resultRpcData.setStatus(status);  // 成功
        resultRpcData.setPath(rpcData.getPath());
        resultRpcData.setVersion(version);
        resultRpcData.setAuthCode(authCode);
        RpcExtData rpcExtData = new RpcExtData();
        rpcExtData.getExtDataMap().put(RpcExtData.LOCAL_SEQ, rpcData.getExtData().getExtDataMap().get(RpcExtData.LOCAL_SEQ));
        resultRpcData.setExtData(rpcExtData);
        resultRpcData.setData(resultObj);
        return resultRpcData;
    }
}
